package rpc.core.plugin.impl;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.StringRedisTemplate;
import rpc.core.FlowMap;
import rpc.core.plugin.ServicePlugin;
import rpc.core.plugin.impl.proxy.RedisServiceProxy;
import rpc.core.plugin.impl.proxy.RedisStringRedisTemplateServiceProxy;
import rpc.core.plugin.impl.proxy.ServiceProxy;

/*********
 *
 *
 *@author 197
 *@date 2020/8/4 12:40
 *
 **/


public class RedisServicePlugin implements ServicePlugin {

  private static final Logger logger = LoggerFactory.getLogger(RedisServicePlugin.class);
  private final Map<String, ServiceProxy> flowInfoHashMap = new HashMap<>();
  private final ApplicationContext context;
  private final Environment environment;
  private int queueSize = 1000;
  private int threads = 1;
  private int maxThreads = 4;
  private ThreadPoolExecutor executor;
  private String beanName = "redis";

  public RedisServicePlugin(ApplicationContext context, Environment environment) {
    this.context = context;
    this.environment = environment;
  }

  @Override
  public void init() {

    threads = intConfig(environment, "rpc." + beanName + ".threads", 8);
    maxThreads = intConfig(environment, "rpc." + beanName + ".maxthreads", 8);
    queueSize = intConfig(environment, "rpc." + beanName + ".queue", 10000);

    executor = new ThreadPoolExecutor(threads, maxThreads, 0, TimeUnit.SECONDS,
        new ArrayBlockingQueue<>(queueSize), new ThreadFactoryBuilder().setNameFormat("pool_redis_%d").build());

    StringRedisTemplate redisTemplate = this.context.getBean(beanName, StringRedisTemplate.class);

    RedisStringRedisTemplateServiceProxy redisServiceProxy = new RedisStringRedisTemplateServiceProxy(redisTemplate);
    Method[] declaredMethods = RedisStringRedisTemplateServiceProxy.class.getDeclaredMethods();
    Arrays.stream(declaredMethods).forEach(m -> {
      String flowName = beanName + "." + m.getName().toLowerCase();
      flowInfoHashMap
          .put(flowName, new RedisServiceProxy().setRedisServiceProxy(redisServiceProxy).setMethod(m).setExecutorService(executor));
    });
    executor.prestartCoreThread();
  }

  @Override
  public Set<String> getFlowName() {
    return null;
  }

  @Override
  public void close() {
    if (Objects.nonNull(executor)) {
      executor.shutdown();
    }
  }

  @Override
  public Map<String, ServiceProxy> getFlowServiceProxy() {
    return this.flowInfoHashMap;
  }

  public String getBeanName() {
    return StringUtils.firstNonBlank(beanName, "redis").toLowerCase();
  }

  public RedisServicePlugin setBeanName(String beanName) {
    this.beanName = beanName;
    return this;
  }

  @Override
  public Map<String, Object> dump() {
    FlowMap flowMap = new FlowMap();
    flowMap.append("threads", threads);
    flowMap.append("maxThreads", maxThreads);
    return flowMap.toMap();
  }
}
